package com.atguigu.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.Date;


/**
 * 该类表示mq的消费者
 * 消费者指定交换机，接受消息，并处理
 */
@Configuration//使用component也可以，只要保证容器在初始化时，创建该对象就好了
public class MqListener {

    /**
     * 接受消息，手动ack确认。如果发生业务异常，我们重新再试一次，还是异常，丢弃消息
     */
    //通过@RabbitListener注解 标注的方法就是一个mq的消费者,该注解可以配置连接的交换机和队列
    @RabbitListener(
            //queues=   如果已经有创建的交换机和绑定到交换机上的队列，我们直接通过queues去指定队列名称，就可以绑定成功了
            bindings = @QueueBinding(//如果没有创建的交换机和队列，我们在这里初始化交换机、队列、绑定关系
                    //@Queue代表初始化队列，name表示队列名称，durable表示队列持久化(mq重启了，队列和里面的消息 依然存在)
                    value = @Queue(name = "spring_test1_queue", durable = "true"),
                    //@Exchange表示初始化交换机 ignoreDeclarationExceptions表示忽略重复声明的异常。type=ExchangeTypes.TOPIC,才能使用通配符。当mq重启，之前创建的交换机存在，我们又创建了一个新的同名交换机，会报异常，忽略这个异常，把之前的交换机覆盖掉
                    exchange = @Exchange(name = "spring.test1.exchange", type = ExchangeTypes.TOPIC, durable = "true", ignoreDeclarationExceptions = "true"),
                    //当前队列绑定到交换机上时，监听消息的路由key(以后交换机中的消息路由key和本次配置的一样才会将消息路由给当前的队列处理)
                    key = "#"//路由key的匹配规则：1.路由key如果完全相同可以进行匹配。2.路由key可以通过通配符进行匹配#(通配所有) *(通配一层，如a.*.*)
            )
    )
    //参数1：接收到的消息内容 参数2：代表消息对象(包含消息的唯一标记) 参数3：代表通道 用来确认消息时使用
    public void test1Consumer(String msg, Message message, Channel channel) {//这个参数顺序无所谓，用几个写几个
        try {
            System.out.println("消费者接受到消息" + msg);
            //接收消息发生异常会发生什么呢，application.yml配置里面了ack，
            // auto表示不断重新发送消息，消息队列里面会一直显示有一条消息未被确认。
            // none表示无所谓，直接确认，mq直接将消息队列中的消息删除，消息队列里面无未确认消息，
            // manuel表示手动确认，用私信队列和延迟队列 解决手动确认时，异常消息的问题，详见catch里的处理
            int i = 1 / 0;//手动制造业务异常

            //手动确认消息被消费(手动ack,即yml文件中acknowledge-mode: manual)时，
            //只有此方法调用成功 mq才会将消息队列中的消息删除，表示消费成功，不写该方法，消息队列中一直显示该消息未确认
            //所以这个方法我们一般写在业务下面，业务没有异常，我们才消费
            //当业务出现异常时，执行不到的该行
            //参数1：代表当前消息的唯一标记  参数2：是否批量处理其他的消息ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

        } catch (Exception e) {
            //发生异常了，我们接下来要决定队列里的消息该怎么处理了
            //第一次的话，先让消息重新归队，我们再试一次，如果还是不行，丢弃消息
            e.printStackTrace();
            try {
                if (!message.getMessageProperties().isRedelivered()) {//消息出异常第一次被捕获
                    //不确认消息，让消息重新回到队列中
                    //参数3：消息是否重新归队，true表示重新回到队列
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                } else {//消息第二次尝试后还是出异常
                    //不确认消息，并拒绝消息，丢弃消息
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }

        }
    }


    /**
     * 接受消息，手动ack确认。如果发生业务异常，我们重新再试一次，还是异常，丢弃消息
     * 丢弃的消息使用死信队列接受存储
     */
    @RabbitListener(
            queues = "test2_service_queue"//直接绑定我们在MqConsumerConfig类里面创建的队列就好了
    )
    public void test2DeadConsumer(String msg, Message message, Channel channel) {
        try {
            System.out.println("test2接收到的消息" + msg);
            int i = 1 / 0;//手动制造业务异常
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            try {
                if (!message.getMessageProperties().isRedelivered()) {//消息出异常第一次被捕获
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                } else {//消息第二次尝试后还是出异常
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
            e.printStackTrace();
        }

    }


    /**
     * 接受消息，手动ack确认。如果发生业务异常，我们重新再试一次，还是异常，丢弃消息
     * 丢弃的消息使用死信队列接受存储，
     * 并使用延迟队列,再次进行消费，所以我们还要创建一个延迟队列的消费者
     * 如果这次消费还是无法消费，我们就放弃了，可以将错误打印日志
     * 无论这次消费成功还是失败，死信队列的消息都会被删除
     * 延迟队列的消费者
     */
    @RabbitListener(queues = "test2_delay_queue")
    public void test3DeadAndDelayConsumer(String msg, Message message, Channel channel) {
        try {
            System.out.println("延迟队列者开始消费消息：" + msg + ",消费时间" + new Date());
            //手动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            e.printStackTrace();
            //将错误打印到日志、redis或者数据库表中
        }

    }


}
